/**
 * $version:  0.1 
 * $Date: 2012-03-19 
 *
 * Copyright (C) 2010-2011 Jawa Software. All rights reserved.
 *
 */

package org.jawa.core.cluster;

import java.util.Collection;
import java.util.Queue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.LinkedBlockingQueue;

import org.jawa.core.cache.CacheFactory;
import org.jawa.core.cache.ClusterExternalUtilStrategy;
import org.jawa.core.cache.ExternalizableUtil;
import org.jawa.core.cache.ExternalizableUtilStrategy;
import org.jawa.core.util.JawaGlobals;
import org.jawa.core.util.JawaProperties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 集群管理,包括集群的数据统计.
 * 
 */
public class ClusterManager {

	private static final Logger Log = LoggerFactory.getLogger(ClusterManager.class);

	public static String CLUSTER_PROPERTY_NAME = "cache.clustering.enabled";
	
	private static ExternalizableUtilStrategy serializationStrategy;
	
	/**
	 * 集群事件监听器
	 */
	private static Queue<ClusterEventListener> listeners = new ConcurrentLinkedQueue<ClusterEventListener>();

	/**
	 * 集群事件
	 */
	private static BlockingQueue<Event> events = new LinkedBlockingQueue<Event>(10000);

	static {
		Thread thread = new Thread("ClusterManager events dispatcher") {
			@Override
			public void run() {
				for (;;) {
					try {
						Event event = events.take();
						EventType eventType = event.getType();
						if (eventType == EventType.joined_cluster && event.getMemberName() == null) {
							// 替换本地缓存至集群缓存.
							// 本地缓存数据并不删除.
							CacheFactory.joinedCluster();
						}
						// 处理事件
						for (ClusterEventListener listener : listeners) {
							try {
								switch (eventType) {
								case joined_cluster: {
									if (event.getMemberName() == null) {
										listener.joinedCluster();
									} else {
										listener.joinedCluster(event.getMemberName());
									}
									break;
								}
								case left_cluster: {
									if (event.getMemberName() == null) {
										listener.leftCluster();
									} else {
										listener.leftCluster(event.getMemberName());
									}
									break;
								}
								case marked_senior_cluster_member: {
									listener.markedAsSeniorClusterMember();
									break;
								}
								default:
									break;
								}
							} catch (Exception e) {
								Log.error(e.getMessage(), e);
							}
						}
						// 标记事件处理完成
						event.setProcessed(true);
					} catch (InterruptedException e) {
						Log.warn(e.getMessage(), e);
					} catch (Exception e) {
						Log.error(e.getMessage(), e);
					}
				}
			}
		};
		thread.setDaemon(true);
		thread.start();
	}

	/**
	 * 注册一个监听器来接收事件。
	 * 
	 * @param listener
	 *            监听器.
	 */
	public static void addListener(ClusterEventListener listener) {
		if (listener == null) {
			throw new NullPointerException();
		}
		listeners.add(listener);
	}

	/**
	 * 注销监听器接收事件
	 * 
	 * @param listener
	 *            监听器事件.
	 */
	public static void removeListener(ClusterEventListener listener) {
		listeners.remove(listener);
	}

	/**
	 * 触发事件表明当前JVM现在是群集的一部分。
	 * 
	 * @param asynchronous
	 *            是否在后台触发事件.
	 */
	public static void fireJoinedCluster(boolean asynchronous) {
		try {
			Event event = new Event(EventType.joined_cluster, null);
			events.put(event);
			if (!asynchronous) {
				while (!event.isProcessed()) {
					Thread.sleep(50);
				}
			}
		} catch (InterruptedException e) {
			Log.error(e.getMessage(), e);
		}
	}

	/**
	 * 触发事件表明另一个JVM现在是群集的一部分。
	 * 
	 * @param nodeID
	 *            集群成员标识.
	 * @param asynchronous
	 *            是否在后台触发事件.
	 */
	public static void fireJoinedCluster(String memberName, boolean asynchronous) {
		try {
			Event event = new Event(EventType.joined_cluster, memberName);
			events.put(event);
			if (!asynchronous) {
				while (!event.isProcessed()) {
					Thread.sleep(50);
				}
			}
		} catch (InterruptedException e) {
			Log.error(e.getMessage(), e);
		}
	}

	/**
	 * 触发事件表明当前JVM 不再是集群成员.
	 */
	public static void fireLeftCluster() {
		for (ClusterEventListener listener : listeners) {
			try {
				listener.leftCluster();
			} catch (Exception e) {
				Log.error(e.getMessage(), e);
			}
		}
	}

	/**
	 * 触发事件表明别外一 JVM 不再是集群成员.
	 * 
	 * @param nodeID
	 *            集群成员标识.
	 */
	public static void fireLeftCluster(String memberName) {
		try {
			Event event = new Event(EventType.left_cluster, memberName);
			events.put(event);
		} catch (InterruptedException e) {
			Log.error(e.getMessage(), e);
		}
	}

	/**
	 * 触发事件表明此JVM现在是高级群集成员。
	 */
	public static void fireMarkedAsSeniorClusterMember() {
		try {
			events.put(new Event(EventType.marked_senior_cluster_member, null));
		} catch (InterruptedException e) {
			// 忽略
		}
	}

	/**
	 * 初始化集群.
	 */
	private static void initForClustering() {
		serializationStrategy = ExternalizableUtil.getInstance().getStrategy();

		ExternalizableUtil extUtil = ExternalizableUtil.getInstance();
		extUtil.setStrategy(new ClusterExternalUtilStrategy());
		
	}

	/**
	 * 启动集群.
	 */
	public static synchronized void startup() {
		if (isClusteringStarted()) {
			return;
		}
		if (isClusteringEnabled()) {
			CacheFactory.startClustering();
		}
	}

	/**
	 * 停止集群缓存.
	 */
	public static synchronized void shutdown() {
		if (isClusteringStarted()) {
			Log.debug("ClusterManager: 停止集群缓存服务.");
			CacheFactory.stopClustering();
			ExternalizableUtil.getInstance().setStrategy(serializationStrategy);
		}
	}

	/**
	 * 设置集群启用,并尝试启动集群.
	 * 
	 * @param enabled
	 *            if clustering support is enabled.
	 */
	public static void setClusteringEnabled(boolean enabled) {
		if (enabled) {
			if (isClusteringEnabled() && isClusteringStarted()) {
				return;
			}
		} else {
			if (!isClusteringEnabled()) {
				return;
			}
		}
		JawaGlobals.setXMLProperty(CLUSTER_PROPERTY_NAME, Boolean.toString(enabled));
		if (!enabled) {
			shutdown();
		} else {
			// 重新加载属性文件
			JawaProperties.getInstance().init();
			initForClustering();
			startup();
		}
	}

	/**
	 * 返回系统配置文件中是否启用集群支持.
	 * 
	 * @return 是否启用集群.
	 */
	public static boolean isClusteringEnabled() {
		return JawaGlobals.getXMLProperty(CLUSTER_PROPERTY_NAME, false);
	}

	/**
	 *  用于判断本JVM是否存在集群。
	 * 
	 * @return 用于判断本JVM是否存在集群.
	 */
	public static boolean isClusteringAvailable() {
		return CacheFactory.isClusteringAvailable();
	}

	/**
	 * 返回集群是否正在启动，如果集群没能启动或是启动失败返回假。
	 * 
	 * @return 集群是否正在启动.
	 */
	public static boolean isClusteringStarting() {
		return CacheFactory.isClusteringStarting();
	}

	/**
	 * 返回集群是否已经启动。
	 * 
	 * @return 集群是否已经启动.
	 */
	public static boolean isClusteringStarted() {
		return CacheFactory.isClusteringStarted();
	}

	/**
	 * 当前 JVM 是否是一个高级集群成员.
	 * 
	 * @return 当前 JVM 是否是一个高级集群成员.
	 */
	public static boolean isSeniorClusterMember() {
		return CacheFactory.isSeniorClusterMember();
	}

	/**
	 * 返回集群中所有成员的信息.
	 * 
	 * @return 集群中所有成员的信息.
	 */
	public static Collection<ClusterNodeInfo> getNodesInfo() {
		return CacheFactory.getClusterNodesInfo();
	}

	/**
	 * 返回集群的最大节点数量，如果为 0 或 1 代表集群不可以使用。
	 * 
	 * @return 集群的最大节点数量，如果为 0 或 1 代表集群不可以使用。
	 */
	public static int getMaxClusterNodes() {
		return CacheFactory.getMaxClusterNodes();
	}

	/**
	 * 返回集群中高级成员的节点ID.
	 * 
	 * @return 集群中高级成员的节点ID.
	 */
	public static String getSeniorClusterMember() {
		return CacheFactory.getSeniorClusterMemberName();
	}

	private static class Event {
		private EventType type;
		private String memberName;
		private boolean processed;

		public Event(EventType type, String memberName) {
			this.type = type;
			this.memberName = memberName;
		}

		public EventType getType() {
			return type;
		}

		public String getMemberName() {
			return memberName;
		}

		public boolean isProcessed() {
			return processed;
		}

		public void setProcessed(boolean processed) {
			this.processed = processed;
		}

		@Override
		public String toString() {
			return super.toString() + " type: " + type;
		}
	}

	/**
	 * 集群事件类型.
	 */
	private enum EventType {

		/**
		 * 当前JVM加入集群.
		 */
		joined_cluster,

		/**
		 * 当前JVM离开集群.
		 */
		left_cluster,

		/**
		 * 当前JVM升级为高级集群成员.
		 */
		marked_senior_cluster_member
	}
}
